package org.niit.service

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

// 离线分析
class MengBatchService {

  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
  val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  spark.sparkContext.setLogLevel("ERROR")

  // 总方法：
  def dataAnalysis(): Unit = {


    val data = spark.read.option("header", "true").csv("input/data_jd.csv")
    //      .as[MengBatchData]


    data.show()

    mengBatch1(data)

    mengBatch2(data)

    mengBatch3(data)

    mengBatch4(data)

    mengBatch5(data)
//
    mengBatch6(data)

  }

  // 需求1： 分析不同操作系统用户的来源渠道分布。
  def mengBatch1(data: DataFrame): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    // 按小时分组并统计观看量（假设每行代表一次观看）
    // 假设已有DataFrame df，包含H(时间), finish(完播), like(点赞)三列
    // 统计不同H下的完播率和点赞率
    data.createOrReplaceTempView("a")
    val analysisResult = spark.sql(
      """
  SELECT
    operative_system,
    source,
    COUNT(user_id) AS user_count
FROM
    a
GROUP BY
    operative_system,
    source
ORDER BY
    operative_system,
    user_count DESC;
  """)

    // 显示结果
    analysisResult.show()

    analysisResult.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/mengBatch1.csv") // 输出目录，内部包含一个 CSV 文件

  }

  // 需求2： 研究各市场级别下用户浏览主页的转化率差异。
  def mengBatch2(data: DataFrame): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    // 按小时分组并统计观看量（假设每行代表一次观看）
    // 假设已有DataFrame df，包含H(时间), finish(完播), like(点赞)三列
    // 统计不同H下的完播率和点赞率
    data.createOrReplaceTempView("a")
    val analysisResult = spark.sql(
      """
  SELECT
    market,  -- 市场级别
    COUNT(DISTINCT user_id) AS total_users,
    SUM(CASE WHEN home_page = 1 THEN 1 ELSE 0 END) AS home_page_users,
    ROUND(
        SUM(CASE WHEN home_page = 1 THEN 1 ELSE 0 END) * 100.0 /
        COUNT(DISTINCT user_id),
        2
    ) AS home_page_conversion_rate
FROM
    a
GROUP BY
    market
ORDER BY
    home_page_conversion_rate DESC;
  """)

    // 显示结果
    analysisResult.show()

    analysisResult.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/mengBatch2.csv") // 输出目录，内部包含一个 CSV 文件

  }


  // 需求3： 比较不同来源渠道用户浏览列表页的转化率。
  def mengBatch3(data: DataFrame): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    // 按小时分组并统计观看量（假设每行代表一次观看）
    // 假设已有DataFrame df，包含H(时间), finish(完播), like(点赞)三列
    // 统计不同H下的完播率和点赞率
    data.createOrReplaceTempView("a")
    val analysisResult = spark.sql(
      """
  SELECT
    source,
    COUNT(DISTINCT user_id) AS total_users,
    SUM(CASE WHEN listing_page = 1 THEN 1 ELSE 0 END) AS listing_page_users,
    ROUND(
        SUM(CASE WHEN listing_page = 1 THEN 1 ELSE 0 END) * 100.0 /
        COUNT(DISTINCT user_id),
        2
    ) AS listing_page_conversion_rate
FROM
    a
GROUP BY
    source
ORDER BY
    listing_page_conversion_rate DESC;
  """)

    // 显示结果
    analysisResult.show()

    analysisResult.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/mengBatch3.csv") // 输出目录，内部包含一个 CSV 文件

  }


  // 需求4： 比较不同来源渠道下不同性别用户的浏览转化率。
  def mengBatch4(data: DataFrame): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    // 按小时分组并统计观看量（假设每行代表一次观看）
    // 假设已有DataFrame df，包含H(时间), finish(完播), like(点赞)三列
    // 统计不同H下的完播率和点赞率
    data.createOrReplaceTempView("a")
    val analysisResult = spark.sql(
      """
SELECT
    source,
    sex,
    COUNT(DISTINCT user_id) AS total_users,
    SUM(CASE WHEN listing_page = 1 THEN 1 ELSE 0 END) AS listing_page_users,
    ROUND(
        SUM(CASE WHEN listing_page = 1 THEN 1 ELSE 0 END) * 100.0 /
        COUNT(DISTINCT user_id),
        2
    ) AS listing_page_conversion_rate
FROM
    a
GROUP BY
    source, sex
ORDER BY
    source, listing_page_conversion_rate DESC;
  """)

    // 显示结果
    analysisResult.show()

    analysisResult.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/mengBatch4.csv") // 输出目录，内部包含一个 CSV 文件

  }


  // 需求5： 分析不同年龄区间在不同设备上的用户分布。
  def mengBatch5(data: DataFrame): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    // 按小时分组并统计观看量（假设每行代表一次观看）
    // 假设已有DataFrame df，包含H(时间), finish(完播), like(点赞)三列
    // 统计不同H下的完播率和点赞率
    data.createOrReplaceTempView("a")
    val analysisResult = spark.sql(
      """
SELECT

    CASE
        WHEN age < 18 THEN '18岁以下'
        WHEN age BETWEEN 18 AND 25 THEN '18-25岁'
        WHEN age BETWEEN 26 AND 35 THEN '26-35岁'
        WHEN age BETWEEN 36 AND 45 THEN '36-45岁'
        WHEN age BETWEEN 46 AND 55 THEN '46-55岁'
        ELSE '55岁以上'
    END AS age_group,
    device AS device_type,
    COUNT(DISTINCT user_id) AS userCount
FROM
    a
GROUP BY
    age_group,
    device
ORDER BY
    age_group,
    device_type;
  """)

    // 显示结果
    analysisResult.show()

    analysisResult.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/mengBatch5.csv") // 输出目录，内部包含一个 CSV 文件

  }



  // 需求6： 分析不同性别用户在不同来源渠道下的支付转化率。
  def mengBatch6(data: DataFrame): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("MengBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    // 按小时分组并统计观看量（假设每行代表一次观看）
    // 假设已有DataFrame df，包含H(时间), finish(完播), like(点赞)三列
    // 统计不同H下的完播率和点赞率
    data.createOrReplaceTempView("a")
    val analysisResult = spark.sql(
      """
SELECT
    sex,
    source,
    ROUND(
        COUNT(CASE WHEN payment_confirmation_page = 1 THEN user_id END) * 1.0 /
        COUNT(CASE WHEN product_page = 1 THEN user_id END),
        2
    ) AS payment_conversion_rate
FROM
    a  -- 替换为实际表名
GROUP BY
    sex,
    source
ORDER BY
    sex,
    source;
  """)

    // 显示结果
    analysisResult.show()

    analysisResult.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/mengBatch6.csv") // 输出目录，内部包含一个 CSV 文件

  }









}
